package com.xinyue.game.service.framwork.receiver;

import java.io.IOException;
import java.util.List;

import javax.annotation.PreDestroy;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.xinyue.game.service.framwork.common.GameServiceConfig;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;


/**
 * 消息队列消费服务
 * 为了支持多实例部署，及用户分散到固定的实例上面，定义网关发送消息时，每一个实例一个topic，每个topic只有一个消息者组
 * ，组名和topic名称一样，比如服务A，B，C,每个服务有两个实例，A1,A2，B1,B2，C1,C2，那么就会产生6个topic
 *
 * @author 王广帅
 * @since 2023/7/2 14:05
 **/
public class RocketMQConsumerService {

    private Logger logger = LoggerFactory.getLogger(RocketMQConsumerService.class);
    @Autowired
    private GameServiceConfig gameServerConfig;

    private DefaultMQPushConsumer pushConsumer;

    /**
     * 启动消息队列的消息者
     */
    public void startRocketMQConsumer(String topic) {
        String endpoints = gameServerConfig.getRocketmqServer();
        String tag = getTag();
        // 一个服务一个分组
        String consumerGroup = topic;
        try {
            pushConsumer = new DefaultMQPushConsumer(consumerGroup);
            pushConsumer.setNamesrvAddr(endpoints);
            pushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            pushConsumer.setMessageModel(MessageModel.CLUSTERING);
            pushConsumer.subscribe(topic, tag);
            pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for (MessageExt messageExt : list) {
                        byte[] byteBuffer = messageExt.getBody();
                        ByteBuf byteBuf = Unpooled.wrappedBuffer(byteBuffer);
                        GameLogicMessageHeader header = new GameLogicMessageHeader();
                        header.readHeader(byteBuf);
                        logger.debug("收到网关消息：{}", header);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            pushConsumer.start();
            logger.info("消息者启动成功，topic: {},tags: {}", topic, tag);
        } catch (Throwable e) {
            logger.error("初始化MQ消费者客户端失败,endpoint: {}", endpoints, e);
            throw new IllegalArgumentException("初始化MQ消费者客户端失败");
        }
    }

    private String getTag() {
        List<String> messageIdList = GameHandlerMappingManager.getInstance().getMessageIdList();
        if (messageIdList.isEmpty()) {
            throw new IllegalArgumentException("没有可消费的tag记录");
        }
        StringBuilder builder = new StringBuilder();
        for (String id : messageIdList) {
            builder.append(id).append("|");
        }
        builder.delete(builder.length() - 1, builder.length());
        return builder.toString();
    }


    @PreDestroy
    public void stopServer() throws IOException {
        if (pushConsumer != null) {
            pushConsumer.shutdown();
            logger.info("关闭消息者成功");
        }
    }
}
